package com.google.code.stalkernet.server;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.google.code.stalkernet.async.Async.Callback;
import com.google.code.stalkernet.async.factory.AsyncFactory;
import com.google.code.stalkernet.async.io.AsyncServerSocketChannel;
import com.google.code.stalkernet.async.io.AsyncSocketChannel;
import com.google.code.stalkernet.node.Node;
import com.google.code.stalkernet.peer.Peer;
import com.google.code.stalkernet.reactor.Reactor;

/**
 * An implementation of {@link Server} interface based on {@link Reactor}.
 *
 * @author Mikhail Vladimirov
 */
public class ReactorServer extends AbstractServer
{
    private final static Logger logger =
        Logger.getLogger (ReactorServer.class.getName ());

    private final Map <SocketAddress, AsyncServerSocketChannel>
        socketAddresses = new HashMap <> ();

    private final Reactor reactor;
    private final AsyncFactory <? super AsyncSocketChannel, ? extends Peer>
        peerFactory;
    private final Node node;

    /**
     * Create new reactor server that uses given {@link Reactor}.
     *
     * @param reactor {@link Reactor} to use
     * @param node {@link Node} to add peers to
     */
    public ReactorServer (
        Reactor reactor,
        AsyncFactory <? super AsyncSocketChannel, ? extends Peer> peerFactory,
        Node node)
    {
        if (reactor == null)
            throw new IllegalArgumentException ("Reactor is null");

        if (peerFactory == null)
            throw new IllegalArgumentException ("Peer factory is null");

        if (node == null)
            throw new IllegalArgumentException ("Node is null");

        this.reactor = reactor;
        this.peerFactory = peerFactory;
        this.node = node;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void addSocketAddress (final SocketAddress socketAddress)
    {
        if (socketAddress == null)
            throw new IllegalArgumentException ("Socet address is null");

        if (socketAddresses.containsKey (socketAddress))
            throw new IllegalStateException (
                "Socket address already added: " + socketAddress);

        AsyncServerSocketChannel asyncServerSocketChannel = null;

        try
        {
            try
            {
                asyncServerSocketChannel =
                    AsyncServerSocketChannel.open (reactor);
            }
            catch (IOException ex)
            {
                logger.log (
                    Level.SEVERE,
                    "Cannot open async server socket channel",
                    ex);
                return;
            }

            try
            {
                asyncServerSocketChannel.bind (socketAddress);
            }
            catch (IOException ex)
            {
                logger.log (
                    Level.SEVERE,
                    "Cannot bind async server socket channel " +
                        asyncServerSocketChannel + " to " + socketAddress,
                    ex);
                return;
            }

            socketAddresses.put (socketAddress, asyncServerSocketChannel);

            asyncServerSocketChannel.accept (
                new AcceptCallback (socketAddress, asyncServerSocketChannel));

            asyncServerSocketChannel = null; // Do not close in finally

            fireSocketAddressAdded (socketAddress);
        }
        finally
        {
            if (asyncServerSocketChannel != null)
            {
                try
                {
                    asyncServerSocketChannel.close ();
                }
                catch (IOException ex)
                {
                    logger.log (
                        Level.SEVERE,
                        "Cannot close async server socket channel: " +
                            asyncServerSocketChannel,
                        ex);
                }
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void removeSocketAddress (SocketAddress socketAddress)
    {
        if (socketAddress == null)
            throw new IllegalArgumentException ("Socket address is null");

        AsyncServerSocketChannel asyncServerSocketChannel =
            socketAddresses.remove (socketAddress);

        if (asyncServerSocketChannel == null)
            throw new IllegalStateException (
                "Socket address does not exist: " + socketAddress);

        try
        {
            asyncServerSocketChannel.close ();
        }
        catch (IOException ex)
        {
            logger.log (
                Level.SEVERE,
                "Cannot close async server socket channel: " +
                    asyncServerSocketChannel,
                ex);
        }

        fireSocketAddressRemoved (socketAddress);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public SocketAddress [] getAllSocketAddresses ()
    {
        return socketAddresses.keySet ().toArray (
            new SocketAddress [socketAddresses.size ()]);
    }

    private class AcceptCallback implements Callback <SocketChannel>
    {
        private final SocketAddress socketAddress;
        private final AsyncServerSocketChannel asyncServerSocketChannel;

        public AcceptCallback (
            SocketAddress socketAddress,
            AsyncServerSocketChannel asyncServerSocketChannel)
        {
            if (socketAddress == null)
                throw new IllegalArgumentException ("Socket address is null");

            if (asyncServerSocketChannel == null)
                throw new IllegalArgumentException (
                    "Async server socket channel is null");

            this.socketAddress = socketAddress;
            this.asyncServerSocketChannel = asyncServerSocketChannel;
        }

        @Override
        public void onSuccess (SocketChannel result)
        {
            try
            {
                final AsyncSocketChannel asyncSocketChannel =
                    new AsyncSocketChannel (result, reactor);

                peerFactory.produce (
                    asyncSocketChannel,
                    new Callback <Peer> ()
                    {
                        @Override
                        public void onSuccess (Peer peer)
                        {
                            if (peer == null)
                                throw new IllegalArgumentException ("Peer is null");

                            node.addPeer (peer);
                        }

                        @Override
                        public void onError (Throwable error)
                        {
                            logger.log (
                                Level.SEVERE,
                                "Cannot create socket channel peer for " +
                                    asyncSocketChannel,
                                error);

                            try
                            {
                                asyncSocketChannel.close ();
                            }
                            catch (IOException exx)
                            {
                                logger.log (
                                    Level.SEVERE,
                                    "Cannot close async socket channel: " +
                                        asyncSocketChannel,
                                    exx);
                            }
                        }
                    });

                result = null;
            }
            catch (IOException ex)
            {
                logger.log (
                    Level.SEVERE,
                    "Cannot create sync socket channel: " + result,
                    ex);
            }
            finally
            {
                if (result != null)
                {
                    try
                    {
                        result.close ();
                    }
                    catch (IOException ex)
                    {
                        logger.log (
                            Level.SEVERE,
                            "Cannot close socket channel: " + result,
                            ex);
                    }
                }
            }

            asyncServerSocketChannel.accept (this);
        }

        @Override
        public void onError (Throwable error)
        {
            logger.log (
                Level.SEVERE,
                "Cannot accept connection on " +
                    asyncServerSocketChannel,
                error);

            removeSocketAddress (socketAddress);
        }
    }
}
